package namedpipe

import (
	"bytes"
	"compress/gzip"
	"encoding/binary"
	"github.com/Microsoft/go-winio"
	"io"
	"log"
	"net"
	"strings"
	"sync"
	"time"
)

const (
	_    = iota
	k    = 1 << (10 * iota)
	m    = 1 << (10 * iota)
	g    = 1 << (10 * iota)
	page = 4 * k
)

// 管道中读写缓冲区的大小
const pipe_buffer_size int32 = 16 * page

// 每次读取数据的大小
const batch_read_size = 1 * page

// 在服务端开始等待客户端连接后多长时间回调启动完成方法
// 增加一定延时以确保服务端就绪后客户端才连接进来
const ready_callback_delay = 120 * time.Millisecond

// 管道名称前缀
const pipe_base_path string = "\\\\.\\pipe\\"

// 向客户端写入内容时加锁，防止并发写的问题
var writeLock sync.Mutex

// 管道的默认配置
// 消息类型为byte，即需要手动处理消息的分隔
// 读写缓冲区大小都为4m
var pipe_conf winio.PipeConfig = winio.PipeConfig{
	MessageMode:      false,
	InputBufferSize:  pipe_buffer_size,
	OutputBufferSize: pipe_buffer_size,
}

// ping帧
var ping_frame []byte = []byte{0, 0, 0, 0}

// 配置的格式
type ServerConf struct {
	ReqCompress     bool
	RespCompress    bool
	RespCompressLev int
}

var defaultServerConf ServerConf = ServerConf{
	ReqCompress:     false,
	RespCompress:    false,
	RespCompressLev: 0,
}

type Writer interface {
	Write(data []byte)
	WriteWithBins(data []byte, binaryDatas [][]byte)
	GetOriginReqId() []byte
}

type connHolder struct {
	conn  net.Conn
	conf  *ServerConf
	reqId []byte
}

type DataHandler interface {
	Handle([]byte, [][]byte, Writer)
}

func CreateFullPipeName(pipename string) string {
	pipename = strings.TrimSpace(pipename)
	if 0 == len(pipename) {
		panic("管道名称不能为空")
	}
	if strings.HasPrefix(pipename, pipe_base_path) {
		return pipename
	}
	return pipe_base_path + pipename
}

func compressData(data []byte, shouldCompress bool, compressLev int) []byte {
	if !shouldCompress {
		return data
	}
	var buffer bytes.Buffer
	writer, _ := gzip.NewWriterLevel(&buffer, compressLev)
	defer writer.Close()
	writer.Write(data)
	writer.Flush()
	writer.Close()
	return buffer.Bytes()
}

func extractData(buf []byte, compressed bool, clone bool) []byte {
	if !compressed {
		if !clone {
			return buf
		}
		return bytes.Clone(buf)
	}

	reader, e := gzip.NewReader(bytes.NewBuffer(buf))
	if nil != e {
		log.Println(e)
		return nil
	}
	defer reader.Close()
	var originBuf bytes.Buffer
	_, e = io.Copy(&originBuf, reader)
	if nil != e {
		log.Println(e)
		return nil
	}
	return originBuf.Bytes()
}

// GetOriginReqId 获取请求id，业务代码一般不需要，只有特定情况下用
func (holder *connHolder) GetOriginReqId() []byte {
	return holder.reqId
}

// Write 向客户端返回结果，不带二进制数据
func (holder *connHolder) Write(data []byte) {
	holder.WriteWithBins(data, nil)
}

// WriteWithBins 向客户端返回结果，带有二进制数据
// 在内部处理数据压缩（如果需要），封装帧格式等内容
func (holder *connHolder) WriteWithBins(data []byte, binaryDatas [][]byte) {
	// 主体数据处理
	factBytes := compressData(data, holder.conf.RespCompress, holder.conf.RespCompressLev)
	payloadSize := len(factBytes)

	// 结果封装
	// 消息体大小 4字节
	// 唯一编号   16字节
	// 消息体内容 n字节
	// 二进制数据个数 4字节
	// 第1个二进制数据大小 4字节
	// 第1个二进制数据内容 n字节
	// 第2个二进制数据大小 4字节
	// 第2个二进制数据内容 n字节
	// ...
	// 计算总数据长度，其中二进制部分要先压缩后再计算长度
	respSize := 4 + 16 + payloadSize + 4
	var compressedBinaryDatas [][]byte
	hasBinaryItems := nil != binaryDatas && 0 < len(binaryDatas)
	binaryCnt := 0
	if hasBinaryItems {
		binaryCnt = len(binaryDatas)
		compressedBinaryDatas = make([][]byte, binaryCnt, binaryCnt)
		for i, binaryData := range binaryDatas {
			compressedData := compressData(binaryData, holder.conf.RespCompress, holder.conf.RespCompressLev)
			respSize += 4 + len(compressedData)
			compressedBinaryDatas[i] = compressedData
		}
	}

	// 写入二进制内容之前的部分
	respBuf := make([]byte, respSize)
	binary.BigEndian.PutUint32(respBuf, uint32(payloadSize))
	copy(respBuf[4:20], holder.reqId)
	copy(respBuf[20:], factBytes)
	binary.BigEndian.PutUint32(respBuf[20+payloadSize:], uint32(binaryCnt))

	// 逐个写入二进制内容：4个字节长度 + n个字节内容
	if hasBinaryItems {
		currOffset := 20 + payloadSize + 4
		for _, binaryData := range compressedBinaryDatas {
			binary.BigEndian.PutUint32(respBuf[currOffset:], uint32(len(binaryData)))
			copy(respBuf[currOffset+4:], binaryData)
			currOffset += 4 + len(binaryData)
		}
	}
	writeBytesExclusive(holder.conn, respBuf)
}

// 向客户端写入结果，通过增加锁以保证写入的原子性
func writeBytesExclusive(conn net.Conn, buf []byte) {
	writeLock.Lock()
	defer writeLock.Unlock()
	conn.Write(buf)
}

// StartNamedPipeServer 启动命名管道并侦听连接，启动成功后会回调startReadyCallback，收到消息到会回调callback.Handle
func StartNamedPipeServer(pipename string, conf *ServerConf, startReadyCallback func(), callback DataHandler) {

	if nil == conf {
		conf = &defaultServerConf
	}

	pipeFullPath := CreateFullPipeName(pipename)
	listener, err := winio.ListenPipe(pipeFullPath, &pipe_conf)
	if nil != err {
		panic(err)
	}
	defer listener.Close()
	log.Printf("Server listening on pipe %v\n", pipeFullPath)

	// 在listener开始等待连接后延迟一定时间回调启动完成函数，以通知客户端可以进行连接
	go func() {
		if nil != startReadyCallback {
			time.Sleep(ready_callback_delay)
			startReadyCallback()
		}
	}()
	for {
		conn, err := listener.Accept()
		if nil != err {
			panic(err)
		}
		// 连接永不超时
		conn.SetReadDeadline(time.Time{})
		go pullData(conn, conf, callback)
	}
}

// 表示不用的处理状态
const (
	st_load_header          = 1 << iota
	st_load_uniqueId        = 1 << iota
	st_load_payload         = 1 << iota
	st_load_bin_header      = 1 << iota
	st_load_bin_item_header = 1 << iota
	st_load_bin_item_cont   = 1 << iota
)

// AccuState 用于累计当前请求的信息
type AccuState struct {
	head struct {
		buf      []byte
		readSize uint32
	}
	uniqueId struct {
		buf      []byte
		readSize uint32
	}
	payload struct {
		buf      []byte
		sumSize  uint32
		readSize uint32
	}
	binHead struct {
		buf      []byte
		readSize uint32
	}
	binCont struct {
		cnt        uint32
		handledCnt uint32
		buf        [][]byte
		itemHead   struct {
			buf      []byte
			readSize uint32
		}
		itemCont struct {
			sumSize  uint32
			readSize uint32
		}
	}
	currState int
	conn      net.Conn
	conf      *ServerConf
	callback  DataHandler
}

type stateHandler interface {
	handle(buf []byte, offset, dataLen uint32) uint32
}

func (st *AccuState) handle(buf []byte, offset, dataLen uint32) uint32 {
	handlerMap := map[int]func(buf []byte, offset, dataLen uint32) uint32{
		st_load_header:          st.readHead,
		st_load_payload:         st.readPayload,
		st_load_bin_header:      st.readBinHead,
		st_load_bin_item_header: st.readBinItemHead,
		st_load_bin_item_cont:   st.readBinItemCont,
		st_load_uniqueId:        st.readUniqueId,
	}
	if fun, ok := handlerMap[st.currState]; ok {
		return fun(buf, offset, dataLen)
	}
	panic("状态有误")
}

// 所有 prepareXXX 函数为处理状态转换的函数，其中分配存储空间，设置当前状态等
func (st *AccuState) prepareReadHead() {
	st.head.readSize = 0
	st.currState = st_load_header
}

func (st *AccuState) prepareReadUniqueId() {
	st.uniqueId.readSize = 0
	st.currState = st_load_uniqueId
}

func (st *AccuState) prepareReadPayload(size uint32) {
	st.payload.buf = make([]byte, size)
	st.payload.readSize = 0
	st.payload.sumSize = size
	st.currState = st_load_payload
}

func (st *AccuState) prepareReadBinHead() {
	st.binHead.readSize = 0
	st.currState = st_load_bin_header
}

func (st *AccuState) prepareReadBinItemHead() {
	st.binCont.itemHead.readSize = 0
	st.currState = st_load_bin_item_header
}

func (st *AccuState) prepareReadBinItemCont(ind uint32, size uint32) {
	st.binCont.itemCont.readSize = 0
	st.binCont.itemCont.sumSize = size
	st.binCont.buf[ind] = make([]byte, size)
	st.currState = st_load_bin_item_cont
}

func (st *AccuState) prepareReadBinCont(cnt uint32) {
	st.binCont.cnt = cnt
	st.binCont.handledCnt = 0
	st.binCont.buf = make([][]byte, cnt, cnt)
	st.prepareReadBinItemHead()
}

// 处理当前状态，此函数内部不直接处理状态的转换，而是通过调用 prepareXXX 方法进行
func (st *AccuState) readHead(buf []byte, offset, dataLen uint32) uint32 {
	readCnt := uint32Min(4-st.head.readSize, dataLen-offset)
	copy(st.head.buf[st.head.readSize:], buf[offset:offset+readCnt])
	offset += readCnt
	st.head.readSize += readCnt

	// 读取完头帧已经有完整4个字节，从中获取体帧长度：
	// 长度为0，则认为是ping帧，回应pong帧，头帧读取长度重新初始化为0；
	// 否则状态改为读取payload
	if 4 == st.head.readSize {
		sumSize := binary.BigEndian.Uint32(st.head.buf)
		if 0 == sumSize {
			st.prepareReadHead()
			writeBytesExclusive(st.conn, ping_frame)
			return offset
		}
		st.prepareReadUniqueId()
	}
	return offset
}

func (st *AccuState) readUniqueId(buf []byte, offset, dataLen uint32) uint32 {
	readCnt := uint32Min(16-st.uniqueId.readSize, dataLen-offset)
	copy(st.uniqueId.buf[st.uniqueId.readSize:], buf[offset:offset+readCnt])
	offset += readCnt
	st.uniqueId.readSize += readCnt

	// uniqueId已经有完整16个字节，状态改为读取payload
	if 16 == st.uniqueId.readSize {
		sumSize := binary.BigEndian.Uint32(st.head.buf)
		st.prepareReadPayload(sumSize)
	}
	return offset
}

func (st *AccuState) readPayload(buf []byte, offset, dataLen uint32) uint32 {
	// 读取payload帧
	readCnt := uint32Min(st.payload.sumSize-st.payload.readSize, dataLen-offset)
	copy(st.payload.buf[st.payload.readSize:], buf[offset:offset+readCnt])
	st.payload.readSize += readCnt
	offset += readCnt

	// 如果体帧读取完成，则状态改为读取二进制头信息
	if st.payload.readSize == st.payload.sumSize {
		st.prepareReadBinHead()
	}
	return offset
}

func (st *AccuState) readBinHead(buf []byte, offset, dataLen uint32) uint32 {
	readCnt := uint32Min(4-st.binHead.readSize, dataLen-offset)
	copy(st.binHead.buf[st.binHead.readSize:], buf[offset:offset+readCnt])
	offset += readCnt
	st.binHead.readSize += readCnt

	// 读够4个字节
	// 如果二进制内容数量为0，即无二进制内容，则状态改为从读取头部，发送响应结果
	// 否则分配二进制内容的空间，并把状态改为读取二进制内容
	if 4 == st.binHead.readSize {
		binaryCnt := binary.BigEndian.Uint32(st.binHead.buf)
		if 0 == binaryCnt {
			st.prepareReadHead()
			handleResp(st)
			return offset
		}
		st.prepareReadBinCont(binaryCnt)
	}
	return offset
}

func (st *AccuState) readBinItemHead(buf []byte, offset, dataLen uint32) uint32 {
	readCnt := uint32Min(4-st.binCont.itemHead.readSize, dataLen-offset)
	copy(st.binCont.itemHead.buf[st.binCont.itemHead.readSize:], buf[offset:offset+readCnt])
	offset += readCnt
	st.binCont.itemHead.readSize += readCnt

	if 4 == st.binCont.itemHead.readSize {
		size := binary.BigEndian.Uint32(st.binCont.itemHead.buf)
		st.prepareReadBinItemCont(st.binCont.handledCnt, size)
	}
	return offset
}

func (st *AccuState) readBinItemCont(buf []byte, offset, dataLen uint32) uint32 {
	readCnt := uint32Min(st.binCont.itemCont.sumSize-st.binCont.itemCont.readSize, dataLen-offset)
	if 0 < readCnt {
		copy(st.binCont.buf[st.binCont.handledCnt][st.binCont.itemCont.readSize:], buf[offset:offset+readCnt])
		offset += readCnt
		st.binCont.itemCont.readSize += readCnt
	}
	if st.binCont.itemCont.readSize == st.binCont.itemCont.sumSize {
		st.binCont.handledCnt++
		if st.binCont.handledCnt < st.binCont.cnt {
			st.prepareReadBinItemHead()
			return offset
		}
		handleResp(st)
		st.prepareReadHead()
	}
	return offset
}

func newAccuState(conn net.Conn, conf *ServerConf, callback DataHandler) *AccuState {
	buffer := make([]byte, 28)
	st := &AccuState{}
	st.head.buf = buffer[0:4]
	st.uniqueId.buf = buffer[4:20]
	st.binHead.buf = buffer[20:24]
	st.binCont.itemHead.buf = buffer[24:28]
	st.currState = st_load_header
	st.conn = conn
	st.conf = conf
	st.callback = callback
	return st
}

// pullData 拉取数据，处理半包粘包情况，封装为完整的请求，转到对应处理器
func pullData(conn net.Conn, conf *ServerConf, callback DataHandler) {
	defer conn.Close()
	buf := make([]byte, batch_read_size)
	offset := uint32(0)
	dataLen := uint32(0)

	// 用于累计每条消息的状态
	var accuState stateHandler = newAccuState(conn, conf, callback)

	// 一批一批把数据读到缓冲区
	for {
		n, e := conn.Read(buf)
		if nil != e {
			log.Println(e)
			break
		}
		if 0 == n {
			continue
		}
		offset = 0
		dataLen = uint32(n)

		// 处理当前批次缓冲区中的数据，直到当前位置已到缓冲区末尾，表示本批数据处理完成
		// 由于有半包粘包问题，本批数据处理完成并不表示得到了一个完整的请求信息
		for {
			if offset >= dataLen {
				break
			}
			offset = accuState.handle(buf, offset, dataLen)
		}
	}
}

// handleResp 处理一个完整的请求，处理前按需要进行解压
// 由于回调函数异步执行，传给回调的参数一定要克隆一份，否则主进程处理时可能会覆盖掉
func handleResp(st *AccuState) {
	holder := connHolder{
		conn:  st.conn,
		conf:  st.conf,
		reqId: bytes.Clone(st.uniqueId.buf),
	}
	restoredBuf := extractData(st.payload.buf, st.conf.ReqCompress, true)
	restoredBins := st.binCont.buf
	if nil != st.binCont.buf && len(st.binCont.buf) > 0 {
		restoredBins = make([][]byte, len(st.binCont.buf))
		for ind, bin := range st.binCont.buf {
			restoredBins[ind] = extractData(bin, st.conf.ReqCompress, true)
		}
	}
	go st.callback.Handle(restoredBuf, restoredBins, &holder)
}

func uint32Min(num1, num2 uint32) uint32 {
	if num1 < num2 {
		return num1
	}
	return num2
}
